package com.jic.syncdata.service.impl;

import com.jic.syncdata.manager.*;
import com.jic.syncdata.service.SyncDataService;
import com.jic.syncdata.utils.LoadConfig;
import com.mysql.cj.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * Desc:
 *
 * @author wulei19922@gmail.com
 * create:2020/12/1
 **/

public class SyncDataServiceImpl implements SyncDataService {

    List<String> excludeKeys = Arrays.asList("database", "orign", "target");

    private static final Logger log = LoggerFactory.getLogger(SyncDataServiceImpl.class);
    int arrayBlockingQueueSize = 1024;//数据读取最大队列 单词读取一晚万 最高支持10240000数据同步
                                      //处理已同步书据也会使用这个参数
    
    public static void main(String[] args) throws SQLException, ClassNotFoundException {

//        
        //1.同步用户
        new SyncDataServiceImpl().syncPmUserService("t_pm_info",new UserManager(),null);


        //2.商品同步
        new SyncDataServiceImpl().syncProduct("t_product_info",new ProduceManager());


        //3.商户同步
        new SyncDataServiceImpl().syncPmUserService("t_channel_info",new MerchanManager(),null);



        //4.订单同步
        new SyncDataServiceImpl().syncPmUserService("t_order_info",new OrderManager(),null);



        //5.优惠券同步
        new SyncDataServiceImpl().syncPmUserService("t_coupon_info",new CouponManager(),null);

        
        //6.启动历史数据处理线程
        SyncDataServiceImpl syncDataService = new SyncDataServiceImpl();
        syncDataService.storageHashBeenSyncData();
        
        


    }
    
    
    
    public  void addTestData(){
        
        ExecutorService queryExecutorService = Executors.newFixedThreadPool(5);


        for (int i = 0; i < 10; i++) {
         

            Thread thread = new Thread(() -> {

                Connection connection=null;
                try {
                     connection = OrignManager.getConnection();
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }

              
                int n=0;
                try {
                    Statement statement = connection.createStatement();
                    connection.setAutoCommit(false);
                
                    
                    
                    while (n<10_000){
                        String  sql="INSERT INTO sync_data.t_user_info ( usr_unionid, usr_openid, address, " +
                                "id_card, name, head_img, tag, formid, create_time, " +
                                "last_login_time, login_times, coupons, last_used_phone, " +
                                "seen_topic, is_received_gold, phones, f_pay_order_time_backup," +
                                " f_pay_order_id, l_pay_order_id, l_pay_order_time, history_search_words," +
                                " f_instalment_order_id, f_instalment_order_time, ecif_no, last_used_invoice, " +
                                "bind_phone, last_used_email, last_used_qq, status, f_pay_order_time, last_nps_fill_time," +
                                " last_nps_push_time, usr_openid_hj) VALUES ('"+UUID.randomUUID().toString()+"', '12312312211412', " +
                                "'湖北武汉市洪山区', '4228021', '吴磊', 'www.baidu.com', '武汉', 'formid', " +
                                "'2020-12-01 18:32:43', '2020-12-01 18:32:46', 0, '1', '13037131937', '武汉都市'," +
                                " '1', '13037131937', '2020-12-01 18:33:21', '1', '1', '2020-12-01 18:33:26'," +
                                " '1', '1', '2020-12-01 18:33:31', '1', '1', '13037131937', '840136024@qq.com'," +
                                " '840136024@qq.com', '0', '2020-12-01 18:33:55', '2020-12-01 18:33:57', " +
                                "'2020-12-01 18:33:58', '1')";
                        statement.addBatch(sql);
                        n++;
                        
                    }
                    
                    statement.executeBatch();
                    connection.commit();
                    connection.close();
                    
                    
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }

              


            });
            
            queryExecutorService.submit(thread);

        }



        while (!queryExecutorService.isShutdown()) {
            queryExecutorService.shutdown();

        }




    }


    /*单表同步都沿用这个方法*/
    public boolean syncPmUserService(String config, ConnectManager connectManager,String orignsql) throws ClassNotFoundException, SQLException {
        Thread queryThread = null;
        Thread inserThread = null;
        log.info("开始同步用户数据");
        log.info("载入配置信息");
        
        AtomicInteger queryCount= new AtomicInteger();//已经查询条数
        AtomicInteger dealCount= new AtomicInteger();//已处理条数
        Map<String, String> t_pm_user = LoadConfig.load("t_pm_info");
     
        /*先使用单线程同步 每次读取20000条数据*/

        /*初始化5个线程*/
        ExecutorService queryExecutorService = Executors.newFixedThreadPool(5);
        ExecutorService insertExecutorService = Executors.newFixedThreadPool(5);
        //1.加载字段映射
        //2.构建查询语句，插入语句。
        //3.启动5个线程进行批处理 （具体性能测试 看网络情况、数据库连接配置）
        ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(arrayBlockingQueueSize);
        final String[] querysql = {buildQuierySql(t_pm_user), buildInsertSql(t_pm_user)};
        final boolean[] starFlag = {true};
        queryThread = new Thread(() -> {
            long satrtid = 0L;
            int size = 1000;
            Connection orign = null;
            try {
                orign = OrignManager.getConnection();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            }
            String tempsql="";
            while (starFlag[0]) {
                //临时处理  按id 升序排列
                log.info(satrtid+"");
                tempsql = querysql[0] + " where  id >"+satrtid+"   order by id asc  limit " + size;
                ResultSet resultSet = null;
                List<HashMap> datalist = new ArrayList<>();
                try {
                    resultSet = orign.createStatement().executeQuery(tempsql);
                    while (resultSet.next()) {
                        HashMap map = new HashMap();
                        ResultSetMetaData metaData = resultSet.getMetaData();
                        for (int i = 1; i < metaData.getColumnCount() + 1; i++) {
                            String columnName = metaData.getColumnName(i).toLowerCase();
                            if ("VARCHAR".equals(metaData.getColumnTypeName(i))) {
                                map.put(columnName, resultSet.getString(columnName));
                            } else if ("DATETIME".equals(metaData.getColumnTypeName(i))) {
                                map.put(columnName, resultSet.getDate(columnName));
                            } else if ("INT".equals(metaData.getColumnTypeName(i))) {
                                map.put(columnName, resultSet.getInt(columnName));

                            } else {
                                log.info(metaData.getColumnTypeName(i));
                            }
                        }
                        datalist.add(map);
                        //记录最后一条数据id
                    };

                    if(datalist.size()>0){
                        satrtid=Long.parseLong(datalist.get(datalist.size()-1).get("id").toString());
                    }
                 
                    log.info(datalist.size()+" "+size);
                    if (datalist.size() < size) {//查询到最后
                        starFlag[0] = false;
                        log.info("查询结束");
                    }
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
                
                queryCount.addAndGet(datalist.size());
                arrayBlockingQueue.add(datalist);
            }
            try {
                orign.close();
            } catch (SQLException throwables) {
                throwables.printStackTrace();
            }
        });


        inserThread = new Thread(() -> {
            try {
                Connection connection = connectManager.getConnection();
                connection.setAutoCommit(false);
                Statement statement = connection.createStatement();
               
                
                while (starFlag[0] ||arrayBlockingQueue.remainingCapacity() < arrayBlockingQueueSize) {
                    List<HashMap> datalist = (List<HashMap>) arrayBlockingQueue.take();
                    //构建sql
                    String tablename = t_pm_user.get("target");
                    StringBuilder buildersql = new StringBuilder("insert  into  ");
                    buildersql.append(tablename);
                    buildersql.append("(");
                    t_pm_user.keySet().forEach(n -> {
                        if (!excludeKeys.contains(n) && !"*".equals(t_pm_user.get(n))) {
                            buildersql.append(t_pm_user.get(n));
                            buildersql.append(",");
                        }
                    });
                    datalist.forEach(n -> {

                        StringBuilder sub = new StringBuilder(buildersql.substring(0, buildersql.length() - 1));
                        sub.append(" ) values ( ");

                        t_pm_user.keySet().forEach(key -> {
                            if (!excludeKeys.contains(key) && !"*".equals(t_pm_user.get(key))) {
                                Object obj = n.get(key);

                                if (obj instanceof String) {

                                    if (StringUtils.isNullOrEmpty(obj.toString())) {
                                        sub.append("'',");
                                    } else {
                                        sub.append("'" + obj + "',");
                                    }

                                } else if (obj instanceof Date) {

                                    sub.append("str_to_date('" + formatdat((Date) obj) + "','%Y-%m-%d %H:%i:%s'),");
                                } else if (obj instanceof Integer) {

                                    sub.append(obj + ",");
                                } else {
                                    sub.append(obj + ",");
//                                   log.info("其他数据类型 未处理");
                                }

                            }
                        });

                        StringBuilder sub1 = new StringBuilder(sub.substring(0, sub.length() - 1));

                        sub1.append(")");


                        try {
                            statement.addBatch(sub1.toString());
                            
                            
                            //数据新增到历史数据队列 该队列有线程负责同步
                            List<Map> collect = datalist.stream().map(tem -> {
                                Map m = new HashMap();
                                m.put("tableid", tem.get("id"));
                                m.put("tablename", t_pm_user.get("orign"));
                                return m;
                            }).collect(Collectors.toList());

                            hashBeenDate.add(collect);
                            
                        } catch (SQLException throwables) {
                            throwables.printStackTrace();
                        }
                    });
                    
                    
                    log.info(starFlag[0] +"  当前剩余容量： "+arrayBlockingQueue.remainingCapacity() +"  总容量"+arrayBlockingQueueSize);
                    //处理一次执行一次
                    statement.executeBatch();
                    connection.commit();
                    dealCount.addAndGet(datalist.size());
                    
                    log.info("当前成功执行条数："+dealCount.get());
                }
                connection.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ClassNotFoundException e) {
                e.printStackTrace();
            } catch (SQLException throwables) {
                throwables.printStackTrace();
            }


        });


        queryThread.start();
        inserThread.start();


        while (!queryExecutorService.isShutdown()) {
            queryExecutorService.shutdown();

        }
        
        return false;
    }

    @Override
    public boolean syncProduct(String config, ConnectManager connectManager) {
        /*商品数据同步 需要多表联合查询   这里传入*/
        String sql="from t_product_info  info\n" +   //商品详情
                "inner   join t_product_detail detail  on detail.prod_code=info.prod_code \n" +//商品详情
                "left   join t_product_storage tps on  tps.prod_code=info.prod_code \n" + //库存
                "inner   join t_channel_info  channel on channel.channel_type=info.channel_type"; //商户


        try {
            syncPmUserService(config,connectManager,sql);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException throwables) {
            log.error("商户数据同步异常："+throwables.getMessage());
        }


        return false;
    }

    @Override
    public boolean syncSeller(String config, ConnectManager connectManager) {
        return false;
    }

    @Override
    public boolean syncOrder(String config, ConnectManager connectManager) {
        return false;
    }

    @Override
    public boolean syncCoupon(String config, ConnectManager connectManager) {
        return false;
    }


    public String formatdat(Date date) {


        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:dd");


        return sdf.format(date);


    }

    /*构建查询sql语句*/

    public String buildQuierySql(Map<String, String> config) {

        //
        if (config == null) {

            log.info("配置错误");
        }
        String tablename = config.get("orign");

        StringBuilder buildersql = new StringBuilder("select ");

        config.keySet().forEach(n -> {
            if (!excludeKeys.contains(n)) {
                buildersql.append(n);
                buildersql.append(",");
            }
        });
        StringBuilder sub = new StringBuilder(buildersql.substring(0, buildersql.length() - 1));
        sub.append(" from ");
        sub.append(tablename);

        return sub.toString();

    }

    private String buildInsertSql(Map<String, String> config) {

        //
        if (config == null) {

            log.info("配置错误");
        }
        String tablename = config.get("target");

        StringBuilder buildersql = new StringBuilder("insert  ");
        buildersql.append(tablename);
        buildersql.append("(");

        config.keySet().forEach(n -> {
            if (!excludeKeys.contains(n) && !"*".equals(config.get(n))) {
                buildersql.append(config.get(n));
                buildersql.append(",");
            }
        });
        StringBuilder sub = new StringBuilder(buildersql.substring(0, buildersql.length() - 1));
        sub.append(" values ");
        return sub.toString();


    }



    
    
    ArrayBlockingQueue hashBeenDate = new ArrayBlockingQueue(arrayBlockingQueueSize*5);
    
    boolean insertTaskDown=false;
    //存储已同步数据
    public  void    storageHashBeenSyncData(){

        new  Thread(()->{
            Map map =new HashMap();
            map.put("tablename","SYNC_DATA_HIS");
            map.put("tableid","1");
            this.hashBeenDate.add(Arrays.asList(map));
        }).start();
        

        ExecutorService executorService = Executors.newFixedThreadPool(2);//初始化参数根据实际性能调节
        
        executorService.submit(new Thread(()->{
            Connection connection = null;

           
            try {
                connection = OrignManager.getConnection();
                connection.setAutoCommit(false);
            } catch (ClassNotFoundException e) {
                log.error("获取连接失败："+e.getMessage());
            } catch (SQLException throwables) {
                throwables.printStackTrace();
            }

            while (!insertTaskDown||hashBeenDate.remainingCapacity()<arrayBlockingQueueSize){
                
               try {
                   List<Map> take = (List<Map>)hashBeenDate.take();
                   String insertSql="insert into SYNC_DATA_HIS(tablename,tableid,syncdate)values('%s','%s',sysdate()) ";
                   Statement statement = connection.createStatement();
                   take.forEach(tem->{
                       String sql = String.format(insertSql, tem.get("tablename"), tem.get("tableid"));
                       try {
                           System.out.println(sql);
                           statement.addBatch(sql);
                       } catch (SQLException throwables) {
                           throwables.printStackTrace();
                       }


                   });

                   statement.executeBatch();
                   connection.commit();
               } catch (InterruptedException e) {
                   e.printStackTrace();
               } catch (SQLException throwables) {
                   throwables.printStackTrace();
               }


            }


            try {
                connection.close();
            } catch (SQLException throwables) {
                throwables.printStackTrace();
            }
        }));

        while (!executorService.isShutdown()){

            log.info("历史数据同步完成");
            executorService.shutdown();
        }
        

    }

 
}
